package com.fanxl.flink.project.app;

import com.alibaba.fastjson.JSON;
import com.fanxl.flink.project.domain.Access;
import com.fanxl.flink.project.domain.AccessV2;
import com.fanxl.flink.project.udf.AccessKafkaDeserializationSchema;
import com.fanxl.flink.project.udf.GaoDeLocationMapFunctionV2;
import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Objects;

/**
 * @description:
 * @author: fanxiaole
 * @date: 2022/3/20 23:35
 */
public class ClickhouseAppV2 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        KafkaSource<Tuple2<String, String>> source = KafkaSource.<Tuple2<String, String>>builder()
                .setBootstrapServers("10.215.7.21:9092,10.215.7.23:9092,10.215.7.24:9092")
                .setTopics("flink_test")
                .setGroupId("test_group")
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setDeserializer(new AccessKafkaDeserializationSchema())
                .setProperty("enable.auto.commit", "false")
                .build();

        FastDateFormat format = FastDateFormat.getInstance("yyyyMMdd-HH");

        DataStreamSource<Tuple2<String, String>> streamSource = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");
        SingleOutputStreamOperator<AccessV2> cleanStream = streamSource.map(new MapFunction<Tuple2<String, String>, AccessV2>() {
            @Override
            public AccessV2 map(Tuple2<String, String> value) throws Exception {
                try {
                    AccessV2 access = JSON.parseObject(value.f1, AccessV2.class);
                    access.id = value.f0;
                    String[] split = format.format(access.time).split("-");
                    access.day = split[0];
                    access.hour = split[1];
                    access.ts = System.currentTimeMillis();
                    return access;
                } catch (Exception e) {
                    e.printStackTrace();
                }
                return null;
            }
        }).filter(x -> x != null)
                .map(new GaoDeLocationMapFunctionV2()).name("数据清洗");
        cleanStream.addSink(JdbcSink.sink("insert into access_log values(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
                (pstmt, x) -> {
                    pstmt.setString(1, x.id);
                    pstmt.setString(2, x.deviceType);
                    pstmt.setString(3, x.uid);
                    pstmt.setString(4, x.os);
                    pstmt.setString(5, x.ip);
                    pstmt.setInt(6, x.nu);
                    pstmt.setString(7, x.channel);
                    pstmt.setLong(8, x.time);
                    pstmt.setString(9, x.event);
                    pstmt.setString(10, x.net);
                    pstmt.setString(11, x.device);
                    pstmt.setString(12, x.version);
                    pstmt.setString(13, x.province);
                    pstmt.setString(14, x.city);
                    pstmt.setString(15, x.day);
                    pstmt.setString(16, x.hour);
                    pstmt.setLong(17, x.ts);
                },
                JdbcExecutionOptions.builder().withBatchSize(50).withBatchIntervalMs(4000).build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
        .withUrl("jdbc:clickhouse://10.215.7.21:8123/flink_study")
        .withDriverName("ru.yandex.clickhouse.ClickHouseDriver")
        .build())).name("CH入库");
        env.execute("ClickhouseAppV1");
    }
}
